package com.hsh.disruptor;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.TimeoutHandler;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class LongEventHandler implements EventHandler<LongEvent>, TimeoutHandler {

    private AtomicBoolean consumerTimeout = new AtomicBoolean(false);

    private int BATCH_DURATION_MS = 2000;
    private int BATCH_SIZE = 5000;
    private List<Long> batch = new ArrayList<>(BATCH_SIZE);
    private AtomicBoolean startConsumer = new AtomicBoolean(false);
    private ReentrantLock lock;
    private Condition notEmpty;

    private long maxBatchEndTime = System.currentTimeMillis() + BATCH_DURATION_MS;

    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws InterruptedException {
//            TimeUnit.SECONDS.sleep(1);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

        batch.add(event.getValue());

        if (batch.size() >= BATCH_SIZE || System.currentTimeMillis() > maxBatchEndTime) {
            Thread.sleep(200);
            System.out.println("process batch : " + batch);
            batch.clear();
            maxBatchEndTime = System.currentTimeMillis() + BATCH_DURATION_MS;
        }
    }

    private void start() {
        lock = new ReentrantLock();
        notEmpty = lock.newCondition();

        final ExecutorService executorService = Executors.newFixedThreadPool(1);
        executorService.execute(new Runnable() {
            @Override
            public void run() {

                for (; ; ) {
                    if (batch.size() >= BATCH_SIZE || System.currentTimeMillis() > maxBatchEndTime) {
                        System.out.println("process batch : " + batch);
                        batch.clear();
                        maxBatchEndTime = System.currentTimeMillis() + BATCH_DURATION_MS;
                    } else {
                        try {
                            notEmpty.await(1000, TimeUnit.MILLISECONDS);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });
    }

    @Override
    public void onTimeout(long sequence) {
        //没有数据消费
        if (!batch.isEmpty() && System.currentTimeMillis() > maxBatchEndTime) {
            System.out.println("process last batch : " + batch);
            batch.clear();
            maxBatchEndTime = System.currentTimeMillis() + BATCH_DURATION_MS;
        }
        System.out.println("timeout sequence" + sequence);
    }


}
